package com.wuwangfu.join;

import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @Author jcshen
 * @Date 2023-02-27
 * @PackageName:com.wuwangfu.join
 * @ClassName: TumbleWindowRightOuterJoin
 * @Description:
 * @Version 1.0.0
 * <p>
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/joining/#tumbling-window-join
 */
public class TumbleWindowRightOuterJoin {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<String> left = env.socketTextStream("localhost", 8888);
        DataStreamSource<String> right = env.socketTextStream("localhost", 9999);
        //生成watermark
        SingleOutputStreamOperator<String> leftWatermark = left.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(String element) {
                return Long.parseLong(element.split(",")[0]);
            }
        });
        SingleOutputStreamOperator<String> rightWatermark = right.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(String element) {
                return Long.parseLong(element.split(",")[0]);
            }
        });
        //map
        SingleOutputStreamOperator<Tuple3<Long, String, String>> leftmaped = leftWatermark.map(new MapFunction<String, Tuple3<Long, String, String>>() {
            @Override
            public Tuple3<Long, String, String> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple3.of(Long.parseLong(fields[0]), fields[1], fields[2]);
            }
        });
        SingleOutputStreamOperator<Tuple3<Long, String, String>> rightmaped = rightWatermark.map(new MapFunction<String, Tuple3<Long, String, String>>() {
            @Override
            public Tuple3<Long, String, String> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple3.of(Long.parseLong(fields[0]), fields[1], fields[2]);
            }
        });
        //right join
        leftmaped.coGroup(rightmaped)
                .where(t -> t.f1)
                .equalTo(t -> t.f1)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new CoGroupFunction<Tuple3<Long, String, String>, Tuple3<Long, String, String>, Tuple6<Long, String, String, Long, String, String>>() {
                    @Override
                    public void coGroup(Iterable<Tuple3<Long, String, String>> first, Iterable<Tuple3<Long, String, String>> second, Collector<Tuple6<Long, String, String, Long, String, String>> out) throws Exception {
                        //循环右流的数据，如果有数据说明触发窗口时右流中有数据
                        for (Tuple3<Long, String, String> ts : second) {
                            boolean flag = false;
                            //循环左流的数据，如果有数据说明触发窗口时左流中有数据，即join上流
                            for (Tuple3<Long, String, String> tf : first) {
                                //返回两个流join上的数据
                                out.collect(Tuple6.of(tf.f0, tf.f1, tf.f2, ts.f0, ts.f1, ts.f2));
                                flag = true;
                            }
                            //如果没有join上，只返回右流的数据
                            if (!flag) {
                                out.collect(Tuple6.of(null, null, null, ts.f0, ts.f1, ts.f2));
                            }
                        }
                    }
                }).print();

        env.execute();
    }
}
